package com.ehotting.mqbd.core.mq.listener;

import com.ehotting.mqbd.core.mq.consumer.action.ConsumerActionAble;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import java.util.List;

@Slf4j
public class MQListenerConcurrently implements MessageListenerConcurrently {

    private static final Logger logger = LoggerFactory.getLogger(MQListenerConcurrently.class);

    private ConsumerActionAble consumerAction;

    public MQListenerConcurrently(ConsumerActionAble consumerAction){
        this.consumerAction = consumerAction;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
       try{
           for(MessageExt msg : msgs){
               consumerAction.invoke(msg);
           }
       }catch (Exception e){
           logger.error("consumeMessage error",e);
       }finally {
           MDC.clear();
       }
       return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
